package cloudy.hbase.dao.impl;

import cloudy.hbase.dao.HBaseDAO;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @Package cloudy.hbase.dao.impl
 * @Description:
 * @Author elwyn
 * @Date 2017/9/24 15:40
 * @Email elonyong@163.com
 */
public class HBaseDAOImpl implements HBaseDAO {
    HConnection hConnection = null;

    public HBaseDAOImpl() {
        Configuration configuration = new Configuration();
        String zk_list = "192.168.241.16,192.168.241.18";
        configuration.set("hbase.zookeeper.quorum", zk_list);
        try {
            hConnection = HConnectionManager.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void save(Put put, String tableName) {
        HTableInterface table = null;
        try {
            table = hConnection.getTable(tableName);
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public void insert(String tableName, String rowKey, String family, String qualifier, String value) {
        HTableInterface table = null;
        try {
            table = hConnection.getTable(tableName);
            Put put = new Put(Bytes.toBytes(rowKey));
            put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public void save(List<Put> puts, String tableName) {
        HTableInterface table = null;
        try {
            table = hConnection.getTable(tableName);
            table.put(puts);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public Result getOneRow(String tableName, String rowKey) {
        HTableInterface table = null;
        Result result = null;
        try {
            table = hConnection.getTable(tableName);
            Get get = new Get(Bytes.toBytes(rowKey));
            result = table.get(get);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return result;
    }

    public List<Result> getRows(String tableName, String rowKey_like) {
        HTableInterface table = null;
        List<Result> results = null;
        try {
            table = hConnection.getTable(tableName);
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes(rowKey_like));
            Scan scan = new Scan();
            scan.setFilter(prefixFilter);
            ResultScanner scanner = table.getScanner(scan);
            results = new ArrayList<Result>();
            for (Result result1 : scanner) {
                results.add(result1);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return results;
    }


    @Override
    public List<Result> getRows(String tableName, String rowKey_like, String cols[]) {
        HTableInterface table = null;
        List<Result> results = null;
        try {
            table = hConnection.getTable(tableName);
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes(rowKey_like));
            Scan scan = new Scan();
            for (String col : cols) {
                scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes(col));
            }
            scan.setFilter(prefixFilter);
            ResultScanner scanner = table.getScanner(scan);
            results = new ArrayList<Result>();
            for (Result result1 : scanner) {
                results.add(result1);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return results;
    }

    @Override
    public List<Result> getRows(String tableName, String startRow, String endRow) {
        HTableInterface table = null;
        List<Result> results = null;
        try {
            table = hConnection.getTable(tableName);
            Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes(startRow));
            scan.setStopRow(Bytes.toBytes(endRow));
            ResultScanner scanner = table.getScanner(scan);
            results = new ArrayList<>();
            for (Result result : scanner) {
                results.add(result);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                if (table != null) {
                    table.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return results;
    }

    public static void main(String[] args) {
        HBaseDAO dao = new HBaseDAOImpl();
      /*  List<Put> puts = new ArrayList<Put>();
        Put put1 = new Put(Bytes.toBytes("testrow1"));
        put1.add(Bytes.toBytes("cf"), Bytes.toBytes("name"), Bytes.toBytes("zhaoliu"));

       // dao.save(put1, "hbase_table");

        put1.add(Bytes.toBytes("cf"), Bytes.toBytes("addr"), Bytes.toBytes("shanghai"));
        puts.add(put1);
        dao.save(puts, "hbase_table");*/
       /* dao.insert("hbase_table", "testrow", "cf", "age", "35");
        dao.insert("hbase_table", "testrow", "cf", "cardid", "126456789");
        dao.insert("hbase_table", "testrow", "cf", "tel", "3512456156");*/
        List<Result> results = dao.getRows("hbase_table", "test", new String[]{"name,age"});
        for (Result result : results) {
            for (Cell cell : result.rawCells()) {
                System.out.println(new String(cell.getRow()));
                System.out.println(new String(cell.getQualifier()));
                System.out.println(new String(cell.getValue()));
            }
        }
    }
}
